package org.example.seata.common.rabbitmq.util;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.json.JSONException;
import cn.hutool.json.JSONObject;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.example.seata.common.core.util.SpringContextHolder;
import org.example.seata.common.rabbitmq.constant.MessageFailDesc;
import org.example.seata.common.rabbitmq.event.MessageFailEvent;
import org.example.seata.common.rabbitmq.message.CorrelationDataEnhance;
import org.example.seata.common.rabbitmq.message.MessageFail;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

import java.io.IOException;
import java.util.List;

/**
 * @author hzq
 * @date 2021/11/25 10:29
 */
@Slf4j
public class MqUtil {

    /**
     * 创建扩展消息
     *
     * @param exchange   交换器
     * @param routingKey 路由键
     * @param message    消息
     * @param msgId      消息id
     * @return 扩展消息
     */
    public static CorrelationDataEnhance createCorrelationDataEnhance(String exchange, String routingKey, Object message, String... msgId) {
        CorrelationDataEnhance correlationDataEnhance = new CorrelationDataEnhance();
        correlationDataEnhance.setExchange(exchange);
        correlationDataEnhance.setRoutingKey(routingKey);
        correlationDataEnhance.setJson(new JSONObject(message).toString());
        if (msgId.length == 1) {
            // 防止消息重复
            // 重发消息时 使用第一次自动生成的消息id
            correlationDataEnhance.setId(msgId[0]);
        }
        return correlationDataEnhance;
    }

    /**
     * 转换和发送相关数据增强
     *
     * @param rabbitTemplate         模板
     * @param correlationDataEnhance 扩展消息
     * @param messageProperties      消息属性
     */
    public static void convertAndSendByCorrelationDataEnhance(RabbitTemplate rabbitTemplate, CorrelationDataEnhance correlationDataEnhance, MessageProperties messageProperties) {
        messageProperties.setCorrelationId(correlationDataEnhance.getId());
        Message message = new Message(correlationDataEnhance.getJson().getBytes(), messageProperties);
        try {
            rabbitTemplate.convertAndSend(correlationDataEnhance.getExchange(), correlationDataEnhance.getRoutingKey(), message, correlationDataEnhance);
        } catch (AmqpException e) {
            MessageFail messageFail = MqUtil.createMessageFail(correlationDataEnhance, MessageFailDesc.CONNECT_MQ, e.getLocalizedMessage());
            SpringContextHolder.publishEvent(new MessageFailEvent(messageFail));
        }
    }

    /**
     * 创建失败消息体
     *
     * @param correlationDataEnhance 扩展消息
     * @param desc                   描述
     * @param cause                  失败报文
     * @return 失败信息
     */
    public static MessageFail createMessageFail(CorrelationDataEnhance correlationDataEnhance, String desc, String cause) {
        return MessageFail.builder()
                .exchange(correlationDataEnhance.getExchange())
                .routingKey(correlationDataEnhance.getRoutingKey())
                .json(correlationDataEnhance.getJson())
                .msgId(correlationDataEnhance.getId())
                .desc(desc)
                .cause(cause)
                .build();
    }

    /**
     * 消息发送到交换器
     * 根据rabbitmq发送过程，消息发送失败的有三种情况会出现：
     * （1）product 连接mq失败，消息没有发送到mq
     * （2）product 连接mq成功，但是发送到exchange失败
     * （3）消息发送到exchange成功，但是路由到queue失败
     * 处理失败消息
     * （1）将失败信息保存到数据库
     */
    public static class CustomConfirmCallback implements RabbitTemplate.ConfirmCallback {

        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            if (!ack) {
                if (correlationData instanceof CorrelationDataEnhance) {
                    // 封装失败消息体
                    CorrelationDataEnhance enhance = (CorrelationDataEnhance) correlationData;
                    MessageFail messageFail = createMessageFail(enhance, MessageFailDesc.SEND_EXCHANGE, cause);
                    // 发布mq消息投递失败事件
                    SpringContextHolder.publishEvent(new MessageFailEvent(messageFail));
                }
            }
        }
    }

    /**
     * @return 自定义交换器确认回调
     */
    public static CustomConfirmCallback buildCustomConfirmCallback() {
        return new CustomConfirmCallback();
    }

    /**
     * 交换器路由到队列
     */
    public static class CustomReturnsCallback implements RabbitTemplate.ReturnsCallback {
        @Override
        public void returnedMessage(ReturnedMessage returned) {
            // 封装失败消息体
            Message message = returned.getMessage();
            CorrelationDataEnhance enhance = new CorrelationDataEnhance();
            enhance.setExchange(returned.getExchange());
            enhance.setRoutingKey(returned.getRoutingKey());
            enhance.setId(message.getMessageProperties().getCorrelationId());
            enhance.setJson(new String(message.getBody()));
            MessageFail messageFail = createMessageFail(enhance, MessageFailDesc.ROUTING_QUEUE, returned.getReplyText());
            // 发布mq消息投递失败事件
            SpringContextHolder.publishEvent(new MessageFailEvent(messageFail));
        }
    }

    /**
     * @return 创建交换器路由对队列失败回调
     */
    public static CustomReturnsCallback buildCustomReturnsCallback() {
        return new CustomReturnsCallback();
    }

    /**
     * 发送消息到mq
     *
     * @param rabbitTemplate 模板
     * @param exchange       交换器
     * @param routingKey     路由键
     * @param message        消息
     * @param msgId          消息id
     */
    public static void sendMessageToRabbitMq(RabbitTemplate rabbitTemplate, String exchange, String routingKey, Object message, String... msgId) {
        sendMessageToRabbitMq(rabbitTemplate, exchange, routingKey, message, new MessageProperties(), msgId);
    }

    /**
     * 发送消息到mq
     *
     * @param rabbitTemplate    模板
     * @param exchange          交换器
     * @param routingKey        路由键
     * @param message           消息
     * @param messageProperties 消息属性
     * @param msgId             消息id
     */
    public static void sendMessageToRabbitMq(RabbitTemplate rabbitTemplate, String exchange, String routingKey, Object message, MessageProperties messageProperties, String... msgId) {
        CorrelationDataEnhance correlationDataEnhance = MqUtil.createCorrelationDataEnhance(exchange, routingKey, message, msgId);
        MqUtil.convertAndSendByCorrelationDataEnhance(rabbitTemplate, correlationDataEnhance, messageProperties);
    }

    /**
     * 批量重发消息
     *
     * @param rabbitTemplate 模板
     * @param messageFails   重发消息列表
     */
    public static void batchResendMessageFails(RabbitTemplate rabbitTemplate, List<MessageFail> messageFails) {
        if (CollUtil.isNotEmpty(messageFails)) {
            for (MessageFail messageFail : messageFails) {
                sendMessageToRabbitMq(rabbitTemplate, messageFail.getExchange(), messageFail.getRoutingKey(), messageFail.getJson(), messageFail.getMsgId());
            }
        }
    }

    /**
     * 解析消息
     *
     * @param message 消息体
     * @param clazz   消息类型
     * @param <T>     泛型
     * @return 消息体
     */
    public static <T> T parseMessage(Message message, Class<T> clazz) {
        return parseMessage(message).toBean(clazz);
    }

    public static JSONObject parseMessage(Message message) {
        String json = new String(message.getBody());
        JSONObject jsonObject;
        try {
            jsonObject = new JSONObject(json);
        } catch (JSONException e) {
            jsonObject = new JSONObject();
            jsonObject.set("message", json);
        }
        return jsonObject;
    }

    /**
     * 恢复消息
     *
     * @param channel 通道
     */
    public static void basicRecover(Channel channel) {
        // 处理失败,重新压入MQ
        try {
            log.error("消费消息失败");
            channel.basicRecover();
        } catch (IOException ex) {
            log.error("恢复消息失败");
        }
    }

    /**
     * 手动确认消息
     *
     * @param message 消息
     * @param channel 通道
     */
    public static void basicAck(Message message, Channel channel) {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            // 手动确认消息
            channel.basicAck(deliveryTag, false);
        } catch (IOException e) {
            basicRecover(channel);
        }
    }

    /**
     * 根据处理结果
     * 手动确认消息
     *
     * @param handlerResult 处理结果
     * @param message       消息
     * @param channel       通道
     */
    public static void basicAck(boolean handlerResult, Message message, Channel channel) {
        if (handlerResult) {
            // 处理成功
            // 手动 ack
            basicAck(message, channel);
        } else {
            // 处理失败 重新将消息压入队列
            basicRecover(channel);
        }
    }
}
